package com.pw.gmall.realtime.app.dws;

import com.pw.gmall.realtime.app.BaseSqlApp;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.KeyWordStats;
import com.pw.gmall.realtime.function.IKAnalyzer;
import com.pw.gmall.realtime.function.KwProduct;
import com.pw.gmall.realtime.utils.FlinkSinkUtils;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: linux_future
 * @since: 2022/3/28
 **/
public class ProductKeyWordApp extends BaseSqlApp {
    public static void main(String[] args) {
        new ProductKeyWordApp().init(2040, 1, "ProductKeyWordApp");
    }

    @Override
    protected void handle(StreamTableEnvironment env) {
        env.executeSql("create table product(" +
                "   stt string, " +
                "   edt string, " +
                "   sku_name string, " +
                "   click_ct bigint, " +
                "   cart_ct bigint, " +
                "   order_ct bigint " +
                ")with(" +
                "   'connector' = 'kafka', " +
                "   'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "   'properties.group.id' = 'ProductKeyWodStatsApp', " +
                "   'topic' = '" + Constant.TOPIC_DWS_PRODUCT_WIDE + "', " +
                "   'scan.startup.mode' = 'earliest-offset', " +
                "   'format' = 'json' " +
                ")");
        Table t1 = env.sqlQuery("select * " +
                " from product" +
                " where click_ct > 0" +
                " or cart_ct > 0" +
                " or order_ct > 0");
        env.createTemporaryView("t1", t1);
        //
        env.createTemporaryFunction("ik_analyzer", IKAnalyzer.class);
        //分词字段sku_name
        Table t2 = env.sqlQuery("select " +
                " stt,edt,word,click_ct,cart_ct,order_ct" +
                " from t1," +
                " lateral table(ik_analyzer(sku_name)) ");
        env.createTemporaryView("t2", t2);

        env.createTemporaryFunction("kw_product", KwProduct.class);

        Table t3 = env.sqlQuery("select " +
                " stt,edt,word keyword,source,ct" +
                " from t2," +
                " lateral table(kw_product(click_ct,cart_ct,order_ct)) as tmp(source,ct) ");

        //t3.execute().print();
        env.createTemporaryView("t3", t3);
        Table t4 = env.sqlQuery("select " +
                " stt,edt,keyword,source," +
                " sum(ct) ct," +
                " unix_timestamp() * 1000 ts" +
                " from t3" +
                " group by stt,edt,keyword,source");

        //t4.execute().print();
        env.toRetractStream(t4, KeyWordStats.class).filter(t -> t.f0)
                .map(t -> t.f1).addSink(FlinkSinkUtils.getClickhouseSink(Constant.CLICKHOUSE_DB,
                        Constant.CLICKHOUSE_TABLE_KEYWORD_STATS, KeyWordStats.class));


    }
}
